/*-------------------------------------------------------------------------
 *
 * blutils.c
 *		布隆索引工具。
 *
 * Portions Copyright (c) 2016-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1990-1993, Regents of the University of California
 *
 * IDENTIFICATION
 *	  contrib/bloom/blutils.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "access/amapi.h"
#include "access/generic_xlog.h"
#include "access/reloptions.h"
#include "bloom.h"
#include "catalog/index.h"
#include "commands/vacuum.h"
#include "miscadmin.h"
#include "storage/bufmgr.h"
#include "storage/freespace.h"
#include "storage/indexfsm.h"
#include "storage/lmgr.h"
#include "utils/memutils.h"

/* 签名处理宏 - 注意假设 i 是 int 类型 */
#define GETWORD(x,i) ( *( (BloomSignatureWord *)(x) + ( (i) / SIGNWORDBITS ) ) )
#define CLRBIT(x,i)   GETWORD(x,i) &= ~( 0x01 << ( (i) % SIGNWORDBITS ) )
#define SETBIT(x,i)   GETWORD(x,i) |=  ( 0x01 << ( (i) % SIGNWORDBITS ) )
#define GETBIT(x,i) ( (GETWORD(x,i) >> ( (i) % SIGNWORDBITS )) & 0x01 )

PG_FUNCTION_INFO_V1(blhandler);

/* 用于 bloom 索引的关系选项类型 */
static relopt_kind bl_relopt_kind;

/* 填充关系选项的解析表 */
static relopt_parse_elt bl_relopt_tab[INDEX_MAX_KEYS + 1];

static int32 fc_myRand(void);
static void fc_mySrand(uint32 fc_seed);

/*
 * 模块初始化函数：初始化关于 Bloom 关系选项的信息。
 *
 * 注意：保持与 makeDefaultBloomOptions() 的同步。
 */
void _PG_init(void)
{
	int			fc_i;
	char		fc_buf[16];

	bl_relopt_kind = add_reloption_kind();

	/* 签名的长度选项 */
	add_int_reloption(bl_relopt_kind, "length",
					  "Length of signature in bits",
					  DEFAULT_BLOOM_LENGTH, 1, MAX_BLOOM_LENGTH,
					  AccessExclusiveLock);
	bl_relopt_tab[0].optname = "length";
	bl_relopt_tab[0].opttype = RELOPT_TYPE_INT;
	bl_relopt_tab[0].offset = offsetof(BloomOptions, bloomLength);

	/* 每个可能索引列的位数：col1, col2, ... */
	for (fc_i = 0; fc_i < INDEX_MAX_KEYS; fc_i++)
	{
		snprintf(fc_buf, sizeof(fc_buf), "col%d", fc_i + 1);
		add_int_reloption(bl_relopt_kind, fc_buf,
						  "Number of bits generated for each index column",
						  DEFAULT_BLOOM_BITS, 1, MAX_BLOOM_BITS,
						  AccessExclusiveLock);
		bl_relopt_tab[fc_i + 1].optname = MemoryContextStrdup(TopMemoryContext,
														   fc_buf);
		bl_relopt_tab[fc_i + 1].opttype = RELOPT_TYPE_INT;
		bl_relopt_tab[fc_i + 1].offset = offsetof(BloomOptions, bitSize[0]) + sizeof(int) * fc_i;
	}
}

/*
 * 构造一组默认的 Bloom 选项。
 */
static BloomOptions *
fc_makeDefaultBloomOptions(void)
{
	BloomOptions *fc_opts;
	int			fc_i;

	fc_opts = (BloomOptions *) palloc0(sizeof(BloomOptions));
	/* 将 DEFAULT_BLOOM_LENGTH 从位数转换为字数 */
	fc_opts->bloomLength = (DEFAULT_BLOOM_LENGTH + SIGNWORDBITS - 1) / SIGNWORDBITS;
	for (fc_i = 0; fc_i < INDEX_MAX_KEYS; fc_i++)
		fc_opts->bitSize[fc_i] = DEFAULT_BLOOM_BITS;
	SET_VARSIZE(fc_opts, sizeof(BloomOptions));
	return fc_opts;
}

/*
 * Bloom 处理函数：返回带有访问方法参数和回调的 IndexAmRoutine。
 */
Datum blhandler(PG_FUNCTION_ARGS)
{
	IndexAmRoutine *fc_amroutine = makeNode(IndexAmRoutine);

	fc_amroutine->amstrategies = BLOOM_NSTRATEGIES;
	fc_amroutine->amsupport = BLOOM_NPROC;
	fc_amroutine->amoptsprocnum = BLOOM_OPTIONS_PROC;
	fc_amroutine->amcanorder = false;
	fc_amroutine->amcanorderbyop = false;
	fc_amroutine->amcanbackward = false;
	fc_amroutine->amcanunique = false;
	fc_amroutine->amcanmulticol = true;
	fc_amroutine->amoptionalkey = true;
	fc_amroutine->amsearcharray = false;
	fc_amroutine->amsearchnulls = false;
	fc_amroutine->amstorage = false;
	fc_amroutine->amclusterable = false;
	fc_amroutine->ampredlocks = false;
	fc_amroutine->amcanparallel = false;
	fc_amroutine->amcaninclude = false;
	fc_amroutine->amusemaintenanceworkmem = false;
	fc_amroutine->amparallelvacuumoptions =
		VACUUM_OPTION_PARALLEL_BULKDEL | VACUUM_OPTION_PARALLEL_CLEANUP;
	fc_amroutine->amkeytype = InvalidOid;

	fc_amroutine->ambuild = blbuild;
	fc_amroutine->ambuildempty = blbuildempty;
	fc_amroutine->aminsert = blinsert;
	fc_amroutine->ambulkdelete = blbulkdelete;
	fc_amroutine->amvacuumcleanup = blvacuumcleanup;
	fc_amroutine->amcanreturn = NULL;
	fc_amroutine->amcostestimate = blcostestimate;
	fc_amroutine->amoptions = bloptions;
	fc_amroutine->amproperty = NULL;
	fc_amroutine->ambuildphasename = NULL;
	fc_amroutine->amvalidate = blvalidate;
	fc_amroutine->amadjustmembers = NULL;
	fc_amroutine->ambeginscan = blbeginscan;
	fc_amroutine->amrescan = blrescan;
	fc_amroutine->amgettuple = NULL;
	fc_amroutine->amgetbitmap = blgetbitmap;
	fc_amroutine->amendscan = blendscan;
	fc_amroutine->ammarkpos = NULL;
	fc_amroutine->amrestrpos = NULL;
	fc_amroutine->amestimateparallelscan = NULL;
	fc_amroutine->aminitparallelscan = NULL;
	fc_amroutine->amparallelrescan = NULL;

	PG_RETURN_POINTER(fc_amroutine);
}

/*
 * 为特定索引填充 BloomState 结构。
 */
void initBloomState(BloomState *fc_state, Relation fc_index)
{
	int			fc_i;

	fc_state->nColumns = fc_index->rd_att->natts;

	/* 为每个属性初始化哈希函数 */
	for (fc_i = 0; fc_i < fc_index->rd_att->natts; fc_i++)
	{
		fmgr_info_copy(&(fc_state->hashFn[fc_i]),
					   index_getprocinfo(fc_index, fc_i + 1, BLOOM_HASH_PROC),
					   CurrentMemoryContext);
		fc_state->collations[fc_i] = fc_index->rd_indcollation[fc_i];
	}

	/* 根据元页的选项初始化 amcache（如有必要） */
	if (!fc_index->rd_amcache)
	{
		Buffer		fc_buffer;
		Page		fc_page;
		BloomMetaPageData *fc_meta;
		BloomOptions *fc_opts;

		fc_opts = MemoryContextAlloc(fc_index->rd_indexcxt, sizeof(BloomOptions));

		fc_buffer = ReadBuffer(fc_index, BLOOM_METAPAGE_BLKNO);
		LockBuffer(fc_buffer, BUFFER_LOCK_SHARE);

		fc_page = BufferGetPage(fc_buffer);

		if (!BloomPageIsMeta(fc_page))
			elog(ERROR, "Relation is not a bloom index");
		fc_meta = BloomPageGetMeta(BufferGetPage(fc_buffer));

		if (fc_meta->magickNumber != BLOOM_MAGICK_NUMBER)
			elog(ERROR, "Relation is not a bloom index");

		*fc_opts = fc_meta->opts;

		UnlockReleaseBuffer(fc_buffer);

		fc_index->rd_amcache = (void *) fc_opts;
	}

	memcpy(&fc_state->opts, fc_index->rd_amcache, sizeof(fc_state->opts));
	fc_state->sizeOfBloomTuple = BLOOMTUPLEHDRSZ +
		sizeof(BloomSignatureWord) * fc_state->opts.bloomLength;
}

/*
 * 从 FreeBSD 复制的随机生成器。在这里使用自己的随机生成器有两个原因：
 *
 * 1）在这种情况下，随机数用于磁盘存储。使用 PostgreSQL 的数字生成器会妨碍它进行所有可能的更改。
 * 2）更改 PostgreSQL 随机生成器的种子将是一个不希望出现的副作用。
 */
static int32 next;

static int32 fc_myRand(void)
{
	/*----------
	 * 计算 x = (7^5 * x) mod (2^31 - 1)
	 * 在不溢出 31 位的情况下：
	 *		(2^31 - 1) = 127773 * (7^5) + 2836
	 * 来源于“随机数生成器：好的难以找到”，
	 * Park 和 Miller，ACM 通讯，卷 31，第 10 期，
	 * 1988 年 10 月，第 1195 页。
	 *----------
	 */
	int32		fc_hi,
				fc_lo,
				fc_x;

	/* 此时必须处于 [1, 0x7ffffffe] 范围内。 */
	fc_hi = next / 127773;
	fc_lo = next % 127773;
	fc_x = 16807 * fc_lo - 2836 * fc_hi;
	if (fc_x < 0)
		fc_x += 0x7fffffff;
	next = fc_x;
	/* 转换为 [0, 0x7ffffffd] 范围。 */
	return (fc_x - 1);
}

static void fc_mySrand(uint32 fc_seed)
{
	next = fc_seed;
	/* 转换为 [1, 0x7ffffffe] 范围。 */
	next = (next % 0x7ffffffe) + 1;
}

/*
 * 将给定值的位添加到签名中。
 */
void signValue(BloomState *fc_state, BloomSignatureWord *fc_sign, Datum fc_value, int fc_attno)
{
	uint32		fc_hashVal;
	int			fc_nBit,
				fc_j;

	/*
	 * 用“列”的数字初始化生成器，以获得新值的“哈希”种子。我们不希望将来自不同列的相同数字映射到相同的位上！
	 */
	fc_mySrand(fc_attno);

	/*
	 * 初始化哈希序列，将我们的值映射到位。不同列中的相同值将因为上面的步骤而映射到不同的位。
	 */
	fc_hashVal = DatumGetInt32(FunctionCall1Coll(&fc_state->hashFn[fc_attno], fc_state->collations[fc_attno], fc_value));
	fc_mySrand(fc_hashVal ^ fc_myRand());

	for (fc_j = 0; fc_j < fc_state->opts.bitSize[fc_attno]; fc_j++)
	{
		/* 防止在 SETBIT 宏中多次评估 */
		fc_nBit = fc_myRand() % (fc_state->opts.bloomLength * SIGNWORDBITS);
		SETBIT(fc_sign, fc_nBit);
	}
}

/*
 * 从值中生成 bloom 元组。
 */
BloomTuple *
BloomFormTuple(BloomState *fc_state, ItemPointer fc_iptr, Datum *fc_values, bool *fc_isnull)
{
	int			fc_i;
	BloomTuple *fc_res = (BloomTuple *) palloc0(fc_state->sizeOfBloomTuple);

	fc_res->heapPtr = *fc_iptr;

	/* 处理每一列的 Bloom */
	for (fc_i = 0; fc_i < fc_state->nColumns; fc_i++)
	{
		/* 跳过 null 值 */
		if (fc_isnull[fc_i])
			continue;

		signValue(fc_state, fc_res->sign, fc_values[fc_i], fc_i);
	}

	return fc_res;
}

/*
 * 将新 Bloom 元组添加到页面。 如果新元组成功添加到页面，则返回 true。 如果它不适合页面，则返回 false。
 */
bool BloomPageAddItem(BloomState *fc_state, Page fc_page, BloomTuple *fc_tuple)
{
	BloomTuple *fc_itup;
	BloomPageOpaque fc_opaque;
	Pointer		fc_ptr;

	/* 我们不应该指向无效的页面 */
	Assert(!PageIsNew(fc_page) && !BloomPageIsDeleted(fc_page));

	/* 新元组是否适合页面？ */
	if (BloomPageGetFreeSpace(fc_state, fc_page) < fc_state->sizeOfBloomTuple)
		return false;

	/* 将新元组复制到页面末尾 */
	fc_opaque = BloomPageGetOpaque(fc_page);
	fc_itup = BloomPageGetTuple(fc_state, fc_page, fc_opaque->maxoff + 1);
	memcpy((Pointer) fc_itup, (Pointer) fc_tuple, fc_state->sizeOfBloomTuple);

	/* 调整 maxoff 和 pd_lower */
	fc_opaque->maxoff++;
	fc_ptr = (Pointer) BloomPageGetTuple(fc_state, fc_page, fc_opaque->maxoff + 1);
	((PageHeader) fc_page)->pd_lower = fc_ptr - fc_page;

	/* 确保我们没有超出可用空间 */
	Assert(((PageHeader) fc_page)->pd_lower <= ((PageHeader) fc_page)->pd_upper);

	return true;
}

/*
 * 分配一个新页面（通过回收或扩展索引文件）
 * 返回的缓冲区已经锁定并获得独占锁定
 * 调用者负责通过调用 BloomInitPage 来初始化页面
 */
Buffer BloomNewBuffer(Relation fc_index)
{
	Buffer		fc_buffer;
	bool		fc_needLock;

	/* 首先，尝试从 FSM 获取一个页面 */
	for (;;)
	{
		BlockNumber fc_blkno = GetFreeIndexPage(fc_index);

		if (fc_blkno == InvalidBlockNumber)
			break;

		fc_buffer = ReadBuffer(fc_index, fc_blkno);

		/*
		 * 我们必须防止其他人已经回收此页面的可能性；
		 * 如果这样，缓冲区可能会被锁定。
		 */
		if (ConditionalLockBuffer(fc_buffer))
		{
			Page		fc_page = BufferGetPage(fc_buffer);

			if (PageIsNew(fc_page))
				return fc_buffer;	/* 可以使用，如果从未初始化 */

			if (BloomPageIsDeleted(fc_page))
				return fc_buffer;	/* 可以使用 */

			LockBuffer(fc_buffer, BUFFER_LOCK_UNLOCK);
		}

		/* 不能使用，因此释放缓冲区并重试 */
		ReleaseBuffer(fc_buffer);
	}

	/* 必须扩展文件 */
	fc_needLock = !RELATION_IS_LOCAL(fc_index);
	if (fc_needLock)
		LockRelationForExtension(fc_index, ExclusiveLock);

	fc_buffer = ReadBuffer(fc_index, P_NEW);
	LockBuffer(fc_buffer, BUFFER_LOCK_EXCLUSIVE);

	if (fc_needLock)
		UnlockRelationForExtension(fc_index, ExclusiveLock);

	return fc_buffer;
}

/*
 * 初始化布隆索引的任何页面。
 */
void BloomInitPage(Page fc_page, uint16 fc_flags)
{
	BloomPageOpaque fc_opaque;

	PageInit(fc_page, BLCKSZ, sizeof(BloomPageOpaqueData));

	fc_opaque = BloomPageGetOpaque(fc_page);
	fc_opaque->flags = fc_flags;
	fc_opaque->bloom_page_id = BLOOM_PAGE_ID;
}

/*
 * 填充布隆索引的元页面。
 */
void BloomFillMetapage(Relation fc_index, Page fc_metaPage)
{
	BloomOptions *fc_opts;
	BloomMetaPageData *fc_metadata;

	/*
	 * 选择索引的选项。如果已经分配了 reloptions，则使用这些选项，否则创建默认选项。
	 */
	fc_opts = (BloomOptions *) fc_index->rd_options;
	if (!fc_opts)
		fc_opts = fc_makeDefaultBloomOptions();

	/*
	 * 初始化元页面的内容，包括选项的副本，这些选项在索引的生命周期内是固定的。
	 */
	BloomInitPage(fc_metaPage, BLOOM_META);
	fc_metadata = BloomPageGetMeta(fc_metaPage);
	memset(fc_metadata, 0, sizeof(BloomMetaPageData));
	fc_metadata->magickNumber = BLOOM_MAGICK_NUMBER;
	fc_metadata->opts = *fc_opts;
	((PageHeader) fc_metaPage)->pd_lower += sizeof(BloomMetaPageData);

	/* 如果这失败了，可能是 FreeBlockNumberArray 大小计算错误： */
	Assert(((PageHeader) fc_metaPage)->pd_lower <= ((PageHeader) fc_metaPage)->pd_upper);
}

/*
 * 初始化布隆索引的元页面。
 */
void BloomInitMetapage(Relation fc_index)
{
	Buffer		fc_metaBuffer;
	Page		fc_metaPage;
	GenericXLogState *fc_state;

	/*
	 * 创建一个新页面；由于这是第一页，因此它应该与块号 0 (BLOOM_METAPAGE_BLKNO) 关联。
	 */
	fc_metaBuffer = BloomNewBuffer(fc_index);
	Assert(BufferGetBlockNumber(fc_metaBuffer) == BLOOM_METAPAGE_BLKNO);

	/* 初始化元页面的内容 */
	fc_state = GenericXLogStart(fc_index);
	fc_metaPage = GenericXLogRegisterBuffer(fc_state, fc_metaBuffer,
										 GENERIC_XLOG_FULL_IMAGE);
	BloomFillMetapage(fc_index, fc_metaPage);
	GenericXLogFinish(fc_state);

	UnlockReleaseBuffer(fc_metaBuffer);
}

/*
 * 解析布隆索引的 reloptions，生成一个 BloomOptions 结构体。
 */
bytea * bloptions(Datum fc_reloptions, bool fc_validate)
{
	BloomOptions *fc_rdopts;

	/* 解析用户提供的 reloptions */
	fc_rdopts = (BloomOptions *) build_reloptions(fc_reloptions, fc_validate,
											   bl_relopt_kind,
											   sizeof(BloomOptions),
											   bl_relopt_tab,
											   lengthof(bl_relopt_tab));

	/* 将签名长度从位数转换为字数，向上取整 */
	if (fc_rdopts)
		fc_rdopts->bloomLength = (fc_rdopts->bloomLength + SIGNWORDBITS - 1) / SIGNWORDBITS;

	return (bytea *) fc_rdopts;
}
